package com.atguigu.flink.streamapi.source;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * Created by Smexy on 2023/1/12
 *
 *  模拟一个Source，会每隔500ms，产生 一条:
 * { userId:100,behavior:"uninstall","channel":"ZTE",timestamp:xxx  }
 *
 * 其中behavior和channel来自如下枚举:
 *  behavior: "install","uninstall","download","update"
 *  channel:"ZTE","APPLE","HUAWEI","MI","VIVO","OPPO"
 */
public class Demo4_CustomSource
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //整个程序中，只有一个线程，运行new MySource()。
        //希望双倍快乐！
        env.setParallelism(2);

        //获取Source，使用flink自带。都是调用env.xxx.不是自带的，都是调用 env.addSource
        //The parallelism of non parallel operator must be 1.
        env.addSource(new MySource()).setParallelism(3)
           .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    //设置为多并行度，必须是ParallelSourceFunction类型
    public static class MySource implements ParallelSourceFunction<String>
    {
    //是单并行度，并行度不能超过1
   // public static class MySource implements SourceFunction<String>{

        boolean ifContinue = true;
        /*
            如何产生数据

            XXXContext：  环境，上下文。 获取各种希望使用的工具。
         */
        @Override
        public void run(SourceContext<String> ctx) throws Exception {

            String [] behaviors = {"install","uninstall","download","update"};
            String [] channels = {"ZTE","APPLE","HUAWEI","MI","VIVO","OPPO"};

            while (ifContinue){

                //本质就是一个Map,代表一个 {  }
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("userId", RandomUtils.nextInt(1,1000));
                jsonObject.put("behavior", behaviors[RandomUtils.nextInt(0,behaviors.length)]);
                jsonObject.put("channel", channels[RandomUtils.nextInt(0,channels.length)]);
                jsonObject.put("timestamp", System.currentTimeMillis());

                //把数据读入
                ctx.collect(jsonObject.toJSONString());

                Thread.sleep(500);

            }


        }

        //停止读数据，不实现。流式计算，源源不断读数据
        @Override
        public void cancel() {

            ifContinue = false;

        }
    }
}
